package pub.cleangao.memm.transport.netty.codec;


import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;
import pub.cleangao.memm.transport.netty.compress.Compress;
import pub.cleangao.memm.transport.netty.compress.CompressSupplier;
import pub.cleangao.memm.transport.netty.compress.CompressType;
import pub.cleangao.memm.transport.message.MessageType;
import pub.cleangao.memm.transport.message.RpcMessage;
import pub.cleangao.memm.transport.TransportConst;
import pub.cleangao.memm.transport.message.RpcRequest;
import pub.cleangao.memm.transport.message.RpcResponse;
import pub.cleangao.memm.transport.netty.serialize.SerializationType;
import pub.cleangao.memm.transport.netty.serialize.Serializer;
import pub.cleangao.memm.transport.netty.serialize.SerializerSupplier;

import java.util.Arrays;


/**
 * protocol:
 * <p>
 * 0     1     2     3     4        5    6    7    8      9          10      11        12   13    14    15 16
 * +-----+-----+-----+-----+--------+----+----+----+------+-----------+-------+--------+-----+-----+----+---+
 * |   magic   code        |version | full length         |messageType| codec |compress|    RequestId       |
 * +-----------------------+--------+---------------------+-----------+-------+--------+--------------------+
 * |                                                                                                        |
 * |                                         body                                                           |
 * |                                                                                                        |
 * |                                        ... ...                                                         |
 * +--------------------------------------------------------------------------------------------------------+
 * 4B  magic code（魔法数）   1B version（版本）   4B full length（消息长度）    1B messageType（消息类型）
 * 1B compress（压缩类型） 1B codec（序列化类型）    4B  requestId（请求的Id）
 * body（object类型数据）
 *
 * @author clean.gao
 */
@Slf4j
public class MessageDecoder extends LengthFieldBasedFrameDecoder {


    public MessageDecoder() {
        // lengthFieldOffset: magic code is 4B, and version is 1B, and then full length. so value is 5
        // lengthFieldLength: full length is 4B. so value is 4
        // lengthAdjustment: full length include all data and read 9 bytes before, so the left length is (fullLength-9). so values is -9
        // initialBytesToStrip: we will check magic code and version manually, so do not strip any bytes. so values is 0
        this(TransportConst.FRAME_MAX_BYTES, 5, 4, -9, 0);
    }


    public MessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        Object decoded = super.decode(ctx, in);
        if (decoded instanceof ByteBuf frame) {
            if (frame.readableBytes() >= TransportConst.PROTOCOL_HEAD_LENGTH) {
                try {
                    return decodeFrame(frame);
                } catch (Exception e) {
                    log.error("decode frame error: {}", e.getMessage());
                    throw e;
                } finally {
                    frame.release();
                }
            }
        }
        return decoded;
    }


    private Object decodeFrame(ByteBuf in) {
        checkMagicNumber(in);
        checkVersion(in);
        int fullLength = in.readInt();
        byte messageType = in.readByte();
        byte codecType = in.readByte();
        byte compressType = in.readByte();
        int requestId = in.readInt();
        RpcMessage rpcMessage = new RpcMessage();
        rpcMessage.setCodec(codecType);
        rpcMessage.setRequestId(requestId);
        rpcMessage.setMessageType(messageType);
        if (messageType == MessageType.HEARTBEAT_REQUEST.value()) {
            rpcMessage.setData(TransportConst.PING);
            return rpcMessage;
        }
        if (messageType == MessageType.HEARTBEAT_RESPONSE.value()) {
            rpcMessage.setData(TransportConst.PONG);
            return rpcMessage;
        }
        int bodyLength = fullLength - TransportConst.PROTOCOL_HEAD_LENGTH;
        if (bodyLength > 0) {
            byte[] bs = new byte[bodyLength];
            in.readBytes(bs);
            // decompress the bytes
            String compressName = CompressType.getName(compressType);
            if (compressName != null) {
                Compress compress = CompressSupplier.getByType(compressName);
                if (compress != null) {
                    bs = compress.decompress(bs);
                }
            }
            // deserialize the object
            String codecName = SerializationType.getName(codecType);
            Serializer serializer = SerializerSupplier.getByType(codecName);
            if (messageType == MessageType.REQUEST.value()) {
                RpcRequest tmpValue = serializer.deserialize(bs, RpcRequest.class);
                rpcMessage.setData(tmpValue);
            } else {
                RpcResponse<?> tmpValue = serializer.deserialize(bs, RpcResponse.class);
                rpcMessage.setData(tmpValue);
            }
        }
        return rpcMessage;

    }

    private void checkVersion(ByteBuf in) {
        byte version = in.readByte();
        if (version != TransportConst.VERSION) {
            throw new RuntimeException("version isn't compatible" + version);
        }
    }

    private void checkMagicNumber(ByteBuf in) {
        int len = TransportConst.MAGIC_NUMBER.length;
        byte[] tmp = new byte[len];
        in.readBytes(tmp);
        for (int i = 0; i < len; i++) {
            if (tmp[i] != TransportConst.MAGIC_NUMBER[i]) {
                throw new IllegalArgumentException("unknown magic code: " + Arrays.toString(tmp));
            }
        }
    }


}
